package com.wuwangfu.state;


import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @Author: jcshen
 * @Date: 2023-03-07
 * @PackageName: com.wuwangfu.state
 * @ClassName: ValueToListState
 * @Description: ValueState实现ListState
 * @Version: 1.0.0
 *
 * 使用ValueState实现ListState的功能
 * ValueState<List<String>>
 */
public class ValueToListState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000);
        //
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000));
        //
        DataStreamSource<String> line = env.socketTextStream("localhost", 8888);
        //
        SingleOutputStreamOperator<Tuple2<String, String>> maped = line.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String value) throws Exception {

                String[] fields = value.split(",");
                return Tuple2.of(fields[0], fields[1]);
            }
        });
        //
        maped.keyBy(t->t.f0)
                .process(new KeyedProcessFunction<String, Tuple2<String, String>, Tuple2<String, List<String>>>() {
                    //
                    private transient ValueState<List<String>> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //定义一个状态描述器
                        ValueStateDescriptor<List<String>> valueStateDescriptor =
                                new ValueStateDescriptor<>("value-state", TypeInformation.of(new TypeHint<List<String>>() {}));
                        //获取状态
                        valueState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(Tuple2<String, String> input, Context context, Collector<Tuple2<String, List<String>>> collector) throws Exception {
                        //获取输入数据
                        String action = input.f1;
                        //获取状态数据
                        List<String> list = valueState.value();
                        if (list == null){
                            list = new ArrayList<>();
                        }
                        //添加状态
                        list.add(action);
                        //更新状态
                        valueState.update(list);
                        //输出
                        collector.collect(Tuple2.of(input.f0,list));
                    }
                }).print();

        env.execute();
    }
}
